package com.rrd.dw.mr.gzip.kafka;

import com.ucredit.babel.common.model.BaseModel;
import com.ucredit.babel.common.model.RecordType;
import com.ucredit.babel.common.model.qcloud.AccessRiskData;
import com.ucredit.babel.common.model.qcloud.AccessRiskScoreInformation;
import com.ucredit.babel.common.model.qcloud.QCloudAccessRiskRecord;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper.Context;

import java.io.IOException;
import java.util.List;

/**
 * 腾讯云行业准入评估
 *
 * @author szw
 */
public class QCloudAccessRiskRecordParse extends ParseBase {
    private Text val = new Text();
    private Text key = new Text();

    /**
     * 腾讯云行业准入评估
     *
     * @param model
     * @throws IOException
     * @throws InterruptedException
     */

    @Override
    public boolean parseJob(Context context, BaseModel<?> model, String successful) throws IOException, InterruptedException {
        QCloudAccessRiskRecord obj = (QCloudAccessRiskRecord) model;
        AccessRiskData data = obj.getData();

        StringBuilder sb = getBaseInfo(obj, successful);
        StringBuilder qcSb = new StringBuilder(2000);
        StringBuilder riSb = new StringBuilder(2000);
        StringBuilder txSb = getBaseInfo(obj);

        //唯一请求 ID
        qcSb.append(data.getRequestId() == null ? "" : data.getRequestId()).append(CRT001);
        //接口调用成功是否的返回码 code
        qcSb.append(data.getData().getCode() == null ? "" : data.getData().getCode()).append(CRT001);
        //出错消息 message
        qcSb.append(data.getData().getMessage() == null ? "" : data.getData().getMessage()).append(CRT001);
        //查得标志 found
        qcSb.append(data.getData().getFound() == null ? "" : data.getData().getFound()).append(CRT001);
        List<AccessRiskScoreInformation> riskInformation = data.getData().getRiskInformation();
        if (!riskInformation.isEmpty()) {
            for (AccessRiskScoreInformation ri : riskInformation) {
                riSb.delete(0, riSb.length());
                //风险等级 riskCodeValue
                riSb.append(ri.getRiskCodeValue() == null ? "" : ri.getRiskCodeValue()).append(CRT001);
                //风险码 riskCode
                riSb.append(ri.getRiskCode() == null ? "" : ri.getRiskCode()).append(CRT001);
                //版本号 version
                riSb.append(ri.getVersion() == null ? "" : ri.getVersion()).append(CRT001);
                key.set(getKey());
                val.set((sb.toString() + qcSb.toString() + riSb.toString() + txSb.toString()).replace("\n", "").replace("\r", " "));
                context.write(key, val);
            }
        } else {
            riSb.append(CRT001);
            riSb.append(CRT001);
            riSb.append(CRT001);
            key.set(getKey());
            val.set((sb.toString() + qcSb.toString() + riSb.toString() + txSb.toString()).replace("\n", "").replace("\r", " "));
            context.write(key, val);
        }
        return true;
    }

    @Override
    public String getTableName() {
        return "kafka_babel_qcloud_access_risk_inc";
    }

    @Override
    public String getType() {
        return RecordType.QCLOUD_ACCESS_RISK_RECORD.toString();
    }

    @Override
    public String getComment() {
        return "腾讯云行业准入评估";
    }
}
